package com.alison.sparkstream.source

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object E1_stream_nc {

"""
    |我们使用 netcat 工具向 9999 端口不断的发送数据，而程序通过 SparkStreaming 读取端口数据并统计不同单词出现的次数。
    |
    |nc -lk 9999
    |hello spark
    |hello java
    |hello kafka
    |""".stripMargin
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ncExample")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    // 通过监控端口创建 DStream，读进来的数据为一行行
    val lineStreams = streamingContext.socketTextStream("192.168.56.104", 9999)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))
    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)
    // 打印结果
    wordAndCountStreams.print()
    // 启动 StreamingContext
    streamingContext.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    streamingContext.awaitTermination()
  }
}
